package org.poem.lock;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.poem.config.IZkLockConfig;
import org.poem.zk.ZkParent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * zookeeper 分布式锁
 */
public class ZkDistributedLock extends ZkParent implements Lock {
    private final static Logger LOG = LoggerFactory.getLogger(ZkDistributedLock.class);

    private static Map<ZkClient, Map<String, LockKeyChangeListener>> zkDataListener = new HashMap<ZkClient, Map<String, LockKeyChangeListener>>();
    /**
     * 本地锁
     */
    private ReentrantLock localLock = new ReentrantLock(true);
    /**
     * 条件
     */
    private Condition localCondition = localLock.newCondition();
    /**
     *
     */
    private LockKeyChangeListener dataListener;
    /**
     * 锁的第二级目录
     */
    private String lockKeyPath;
    /**
     * 当前锁
     */
    private Object initLock = new Object();
    /**
     * 当前的数据
     */
    private String lockPath;

    /**
     * 写入锁的地址
     */
    private String lockData;
    private Integer port;


    /**
     * 创建分布式锁
     *
     * @param config      zk的配置
     * @param lockKeyPath 锁的二级目录   bi-task
     */
    public ZkDistributedLock(IZkLockConfig config, String lockKeyPath) {
        super(config);
        this.lockKeyPath = lockKeyPath;
        Preconditions.checkArgument(!Strings.isNullOrEmpty(lockKeyPath), "ZkDistributedLock 需指定lockKey");
        Preconditions.checkArgument(!lockKeyPath.contains("/"), "lockKey不可包函/");
        initLockPath();
        if (zkDataListener.get(super.getZkClient()) == null || zkDataListener.get(super.getZkClient()).get(lockKeyPath) == null) {
            synchronized (initLock) {
                String basePath = super.getConfig().getBasePath().replaceFirst("(.*)?/$", "$1");
                if (!super.getZkClient().exists(basePath)) {
                    LOG.debug("分布式锁主路径不存在,准备创建!{}", basePath);
                    super.getZkClient().createPersistent(basePath, "true");
                }
                LOG.debug("分布式锁主路径不存在,已创建!{}", super.getConfig().getBasePath());
                if (!super.getZkClient().exists(basePath + "/" + lockKeyPath)) {
                    LOG.debug("分布式锁二级路径不存在,准备创建!{}", basePath);
                    super.getZkClient().createPersistent(basePath + "/" + lockKeyPath, "true");
                }
                LOG.debug("分布式锁二级路径不存在,已创建!{}", super.getConfig().getBasePath() + lockKeyPath);
                zkDataListener.computeIfAbsent(super.getZkClient(), k -> new HashMap<String, LockKeyChangeListener>());
                if (zkDataListener.get(super.getZkClient()).get(lockKeyPath) == null) {
                    this.dataListener = new LockKeyChangeListener(this);
                    zkDataListener.get(super.getZkClient()).put(lockKeyPath, this.dataListener);
                    super.getZkClient().subscribeDataChanges(lockPath, this.dataListener);
                    LOG.debug("初始化锁路径监听,localPath:{}", this.lockPath);
                }
            }
        }
        this.dataListener = (LockKeyChangeListener) zkDataListener.get(super.getZkClient()).get(lockKeyPath);
        dataListener.addDistributedLock(this);
    }


    /**
     * 创建zk的远程锁
     *
     * @return
     */
    protected boolean createRemoteLock() {
        if (Objects.isNull(lockData)) {
            throw new IllegalArgumentException("lock before set [lockData]");
        }
        boolean res = false;
        String zkPath = lockPath + "/" + lockData;
        try {
            super.getZkClient().createEphemeral(zkPath,  lockData  + "@1@"+port);
            LOG.info("{}-{},获取锁成功!", new Object[]{lockData, lockPath});
            res = true;
        } catch (Exception ex) {
            LOG.info("{}-{},获取锁失败!", new Object[]{lockData, lockPath});
            String remoteLockData = super.getZkClient().readData(zkPath, true);
            if (remoteLockData != null && remoteLockData.startsWith(lockData)) {
                int count = Integer.parseInt(remoteLockData.split("@")[1]) + 1;
                super.getZkClient().writeData(zkPath, lockData + "@" + count + "@"+port);
                LOG.debug("{}-{},获取锁成功,获取次数:{}!", new Object[]{lockData, lockPath, count});
            }
        }
        return res;
    }

    /**
     * 移除锁
     */
    protected void removeRemoteLock() {
        String remoteLockData = super.getZkClient().readData(lockPath + "/" + lockData, true);
        if (remoteLockData != null && remoteLockData.startsWith(lockData)) {
            super.getZkClient().delete(lockPath + "/" + lockData);
            LOG.debug("{}-{},远程锁释放成功!", new Object[]{lockData, lockPath + "/" + lockData});
        }
        localLock.unlock();
    }

    /**
     * 上锁
     */
    @Override
    public void lock() {
        for (; ; ) {
            LOG.debug("{}-{},准备获取锁!", new Object[]{lockData, lockPath});
            localLock.lock();
            boolean res = createRemoteLock();
            if (res) {
                break;
            }
            try {
                LOG.debug("{}-{},获取锁失败,进入等待状态!", new Object[]{lockData, lockPath});
                long waitFlag = localCondition.awaitNanos(TimeUnit.MILLISECONDS.toNanos(super.getConfig().getRetryMaxTime() == null ? 300 : super.getConfig().getRetryMaxTime()));
                if (waitFlag > 0) {
                    LOG.debug("{}-{},获取锁收到信息苏醒,重新获取!", new Object[]{lockData, lockPath});
                } else {
                    LOG.debug("{}-{},获取锁自动苏醒,重新获取!", new Object[]{lockData, lockPath});
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 异常 锁
     *
     * @throws InterruptedException
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        for (; ; ) {
            LOG.debug("{}-{},准备获取锁!", new Object[]{lockData, lockPath});
            localLock.lockInterruptibly();
            boolean res = createRemoteLock();
            if (res) {
                break;
            }
            LOG.debug("{}-{},获取锁失败,进入等待状态!", new Object[]{lockData, lockPath});
            long waitFlag = localCondition.awaitNanos(TimeUnit.MILLISECONDS.toNanos(super.getConfig().getRetryMaxTime() == null ? 300 : super.getConfig().getRetryMaxTime()));
            if (waitFlag > 0) {
                LOG.debug("{}-{},获取锁收到信息苏醒,重新获取!", new Object[]{lockData, lockPath});
            } else {
                LOG.debug("{}-{},获取锁自动苏醒,重新获取!", new Object[]{lockData, lockPath});
            }
        }
    }

    /**
     * 尝试获取锁
     *
     * @return
     */
    @Override
    public boolean tryLock() {
        LOG.debug("{}-{},准备获取锁!", new Object[]{lockData, lockPath});
        boolean res = localLock.tryLock();
        if (res) {
            res = createRemoteLock();
        }
        if (!res) {
            LOG.debug("{}-{},获取锁失败!", new Object[]{lockData, lockPath});
        }
        return res;
    }
    /**
     * 尝试获取锁
     *
     * @return
     */

    /**
     * 尝试锁
     *
     * @param time 等待的时长
     * @param unit 等待的单位
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        long lastTime = System.nanoTime();
        long now, offsetTime;
        boolean res = false;
        for (; ; ) {
            now = System.nanoTime();
            offsetTime = now - lastTime;
            if (unit.toNanos(time) - offsetTime > 0) {
                LOG.debug("{}-{},准备获取锁!", new Object[]{lockData, lockPath});
                res = localLock.tryLock(unit.toNanos(time) - offsetTime, unit);
                if (res) {
                    res = createRemoteLock();
                    if (res) {
                        break;
                    }
                    now = System.nanoTime();
                    offsetTime = unit.convert(now - lastTime, TimeUnit.NANOSECONDS);
                    if (time - offsetTime > 0) {
                        LOG.debug("{}-{},获取锁失败,进入等待状态!", new Object[]{lockData, lockPath});
                        long waitFlag = localCondition.awaitNanos(unit.toNanos(time - offsetTime));
                        if (waitFlag > 0) {
                            LOG.debug("{}-{},获取锁收到信息苏醒,重新获取!", new Object[]{lockData, lockPath});
                        } else {
                            LOG.debug("{}-{},获取锁自动苏醒,重新获取!", new Object[]{lockData, lockPath});
                        }
                    } else {
                        break;
                    }
                } else {
                    break;
                }
            } else {
                break;
            }
        }
        if (!res) {
            LOG.debug("{}-{},获取锁失败!", new Object[]{lockData, lockPath});
        }
        return res;
    }

    /**
     * 解锁
     */
    @Override
    public void unlock() {
        localLock.lock();
        try {
            removeRemoteLock();
        } finally {
            localLock.unlock();
        }
    }

    /**
     * 该方法返回的Condition中的方法，当前仅对本地线程有效.
     * await 仅释放本地锁，远程锁将继续保持直到调用unlock或应用中止
     *
     * @return
     * @see Lock#newCondition()
     */
    @Override
    public Condition newCondition() {
        return localLock.newCondition();
    }

    public String getLockData() {
        return lockData;
    }

    public void setLockData(String lockData) {
        this.lockData = lockData;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    /**
     * 销毁锁
     */
    public void destroy() {
        LOG.debug("销毁锁,{}!", lockPath);
        this.dataListener.removeDistributedLock(this);
        super.destroy();
    }

    /**
     * 初始化锁的路径
     */
    private void initLockPath() {
        if (super.getConfig().getBasePath() == null) {
            lockPath = "/";
        } else if (super.getConfig().getBasePath().endsWith("/")) {
            lockPath = super.getConfig().getBasePath();
        } else {
            lockPath = super.getConfig().getBasePath() + "/";
        }
        lockPath = lockPath + lockKeyPath;
    }

    protected ReentrantLock getLocalLock() {
        return localLock;
    }

    protected Condition getLocalCondition() {
        return localCondition;
    }

    /**
     * zk 节点变化监听
     */
    private class LockKeyChangeListener implements IZkDataListener {
        private Set<ZkDistributedLock> lockSet = new HashSet<ZkDistributedLock>();

        public void addDistributedLock(ZkDistributedLock lock) {
            lockSet.add(lock);
        }

        public void removeDistributedLock(ZkDistributedLock lock) {
            lockSet.remove(lock);
        }

        public LockKeyChangeListener(ZkDistributedLock lock) {
            this.addDistributedLock(lock);
        }

        @Override
        public void handleDataChange(String dataPath, Object data) throws Exception {
            LOG.debug("远程锁发生变更,{}-{}!", new Object[]{lockPath, data});
        }

        /**
         * 线程被释放的通知
         *
         * @param dataPath
         * @throws Exception
         */
        @Override
        public void handleDataDeleted(String dataPath) throws Exception {
            if (Objects.nonNull(lockSet)) {
                for (ZkDistributedLock lock : lockSet) {
                    lock.getLocalLock().lock();
                    try {
                        lock.getLocalCondition().signalAll();
                    } finally {
                        lock.getLocalLock().unlock();
                    }
                }
            }
        }
    }
}
